// Copyright 2020 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: provide query interfaces
// feature: Split history as an independent common module
// feature: support message brief record
// ------------------------------------------------------------------

#ifndef _RTPS_BUILTIN_DISCOVERY_ENDPOINT_EDPUTILS_HPP_
#define _RTPS_BUILTIN_DISCOVERY_ENDPOINT_EDPUTILS_HPP_

#include <edds/rtps/attributes/HistoryAttributes.h>
#include <edds/rtps/attributes/ReaderAttributes.h>
#include <edds/rtps/attributes/WriterAttributes.h>
#include <history/ReaderHistory.h>
#include <history/WriterHistory.h>
#include <edds/rtps/reader/StatefulReader.h>
#include <edds/rtps/writer/StatefulWriter.h>

#include <history/ITopicPayloadPool.h>
#include <history/PoolConfig.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <memory>
#include <string>

#include <edds/rtps/common/EntityId_t.hpp>
#include <history/TopicPayloadPoolRegistry.hpp>

namespace evbs {
namespace ertps {
namespace rtps {

class EDPUtils {
 public:
    using WriterHistoryPair = std::pair<StatefulWriter*, WriterHistory*>;
    using ReaderHistoryPair = std::pair<StatefulReader*, ReaderHistory*>;

    static std::shared_ptr<vbs::common::ITopicPayloadPool> create_payload_pool(const std::string& topic_name,
                                                                               const HistoryAttributes& history_attr,
                                                                               bool is_reader) {
        PoolConfig pool_cfg = PoolConfig::from_history_attributes(history_attr);
        auto pool = TopicPayloadPoolRegistry::get(topic_name, pool_cfg);
        (void)pool->reserve_history(pool_cfg, is_reader);
        return pool;
    }

    static void release_payload_pool(std::shared_ptr<vbs::common::ITopicPayloadPool>& pool,
                                     const HistoryAttributes& history_attr, bool is_reader) {
        if (pool) {
            PoolConfig pool_cfg = PoolConfig::from_history_attributes(history_attr);
            (void)pool->release_history(pool_cfg, is_reader);
            pool.reset();
        }
    }

    static bool create_edp_reader(RTPSParticipantImpl* participant, const std::string& topic_name,
                                  const EntityId_t& entity_id, const HistoryAttributes& history_att,
                                  ReaderAttributes& ratt, ReaderListener* listener,
                                  std::shared_ptr<vbs::common::ITopicPayloadPool>& payload_pool,
                                  ReaderHistoryPair& edp_reader) {
        RTPSReader* raux = nullptr;

        payload_pool = create_payload_pool(topic_name, history_att, true);
        try {
            edp_reader.second = new ReaderHistory(history_att);
        } catch (const foonathan::memory::out_of_memory& e) {  //LCOV_EXCL_START
            logWarning(RTPS_PDP, std::string("catch out_of_memory exception:") << e.what());
            return false;
        }  //LCOV_EXCL_STOP

        bool created =
            participant->createReader(&raux, ratt, payload_pool, edp_reader.second, listener, entity_id, true);

        if (created) {
            edp_reader.first = static_cast<StatefulReader*>(raux);
            edp_reader.first->setTopicName("TopicBuiltinEDP");
            std::stringstream str;
            str << edp_reader.first->getGuid().guidPrefix << "|" << edp_reader.first->getGuid().entityId;
            edp_reader.first->register_peroid_metrics(str.str());
        } else {
            delete (edp_reader.second);
            edp_reader.second = nullptr;
            release_payload_pool(payload_pool, history_att, true);
        }

        return created;
    }

    static bool create_edp_writer(RTPSParticipantImpl* participant, const std::string& topic_name,
                                  const EntityId_t& entity_id, const HistoryAttributes& history_att,
                                  WriterAttributes& watt, WriterListener* listener,
                                  std::shared_ptr<vbs::common::ITopicPayloadPool>& payload_pool,
                                  WriterHistoryPair& edp_writer) {
        RTPSWriter* waux = nullptr;

        payload_pool = create_payload_pool(topic_name, history_att, false);
        edp_writer.second = new WriterHistory(history_att);
        bool created =
            participant->createWriter(&waux, watt, payload_pool, edp_writer.second, listener, entity_id, true);

        if (created) {
            edp_writer.first = dynamic_cast<StatefulWriter*>(waux);
            if (edp_writer.first == nullptr) {
                delete (edp_writer.second);
                edp_writer.second = nullptr;
                release_payload_pool(payload_pool, history_att, false);
                return false;
            }
            edp_writer.first->setTopicName("TopicBuiltinEDP");
            edp_writer.first->setSamplesPerHb(20);
            std::stringstream str;
            str << edp_writer.first->getGuid().guidPrefix << "|" << edp_writer.first->getGuid().entityId;
            edp_writer.first->register_peroid_metrics(str.str());
        } else {
            delete (edp_writer.second);
            edp_writer.second = nullptr;
            release_payload_pool(payload_pool, history_att, false);
        }

        return created;
    }
};

} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */

#endif  // _RTPS_BUILTIN_DISCOVERY_ENDPOINT_EDPUTILS_HPP_
